package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: xyl
 * @Description: 针对企查查t_stockrelation_legal_info表数据映射开发
 * @Date: 2021/12/13 17:51
 */

public class TStockrelationLegalInfo {
    public static void main(String[] args) {
        /*SparkSession spark = SparkSession.builder().appName("t_stockrelation_legal_info").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");

        //获取高管是法人的增量数据
        Dataset<Row>  t_stockrelation_legal_info_temp = spark.sql("select \n" +
                "y.pid,\n" +
                "s.staff_name,\n" +
                "s.staff_type_name,\n" +
                "s.update_time,\n" +
                "s.data_state,\n" +
                "s.eid,\n" +
                "s.staff_type_name,\n" +
                "translate(translate(x.ent_name,'（','('),'）',')') as ent_name ,\n" +
                "x.credit_code \n" +
                "from a_dc_ep_ods.zax_t_staff s\n" +
                "inner join (select * from  a_dc_ep_dwi.zax_t_base where ent_name is not null) x on s.eid = x.uid\n" +
                "inner join (select * from  a_dc_ep_dwi.zax_t_pid where ename is not null and pname is not null ) y on upper(s.staff_name) = upper(y.pname) and x.ent_name = y.ename\n" +
                "where s.data_state != 1  ");
        t_stockrelation_legal_info_temp.createOrReplaceTempView("t_stockrelation_legal_info_temp");

        spark.sql("select a.key_no,a.company_id, translate(translate(a.company_name,'（','('),'）',')') as company_name," +
                "a.oper_key_no,a.oper_name,a.regist_capi,a.regist_capi_value,a.regist_capi_unit,a.province_code,a.province,a.no,a.credit_code,a.status,a.econ_kind" +
                " from (select aa.*,ROW_NUMBER() OVER(PARTITION BY aa.company_name ORDER BY aa.dates DESC,aa.isadd DESC) num from a_dc_ep_ods.t_eci_company aa) a").createOrReplaceTempView("t_eci_company_temp");


        //获取新增，修改，删除高管数据   企业信息数据  -1 删除 0 更新  1 新增
        Dataset<Row>  t_stockrelation_legal_info_begin =spark.sql("select \n" +
                "distinct \n" +
                "(case when y.key_no !=''  and y.key_no is not null  then y.key_no else md5(s.credit_code) end ) as key_no,\n" +
                "(case when y.company_id !='' and y.company_id is not null  then y.company_id else md5(s.credit_code) end ) as company_id,\n" +
                "s.ent_name as company_name,\n" +
                "s.staff_name as name,\n" +
                "s.staff_type_name as job,\n" +
                "(case when x.p_key_no !=''  then x.p_key_no else s.pid  end ) as p_key_no,\n" +
                "date_format(s.update_time,'yyyyMMdd') as dates,\n" +
                "(case when s.data_state ='2'  then '-1' when x.key_no !=''  then '0' else '1' end ) as isadd\n" +
                "from  t_stockrelation_legal_info_temp s\n" +
                "inner join t_eci_company_temp  y ON  trim(s.ent_name)= trim(y.company_name)\n" +
                "left join a_dc_ep_dwi.t_eci_employee x ON y.company_id = x.company_id and trim(x.name) = trim(s.staff_name) " +
                "inner join t_eci_company_temp  t ON  trim(t.oper_key_no)= trim(x.p_key_no) " +
                "left join a_dc_ep_dwi.t_eci_industry z on z.company_id = t.company_id  \n" );

        t_stockrelation_legal_info_begin.createOrReplaceTempView("t_stockrelation_legal_info_begin");

        //获取新增，修改，删除董监高法人    企业信息数据，行业分类数据   -1 删除 0 更新  1 新增
        Dataset<Row>  t_stockrelation_legal_info_end =spark.sql("select \n" +
                "distinct \n" +
                "y.key_no,\n" +
                "y.company_id,\n" +
                "y.company_name,\n" +
                "y.oper_key_no,\n" +
                "y.oper_name,\n" +
                "y.regist_capi,\n" +
                "y.regist_capi_value,\n" +
                "y.regist_capi_unit,\n" +
                "y.province_code,\n" +
                "y.province,\n" +
                "z.industry,\n" +
                "y.no,\n" +
                "y.credit_code,\n" +
                "y.status,\n" +
                "y.econ_kind,\n" +
                "s.dates,\n" +
                "s.isadd\n" +
                "from  t_stockrelation_legal_info_begin s\n" +
                "inner join t_eci_company_temp  y ON  trim(y.oper_key_no)= trim(s.p_key_no)  \n" +
                "left join a_dc_ep_dwi.t_eci_industry z on y.company_id = z.company_id");

        t_stockrelation_legal_info_end.createOrReplaceTempView("t_stockrelation_legal_info_end");

        spark.sql("insert overwrite table a_dc_ep_incr.t_stockrelation_legal_info select * from t_stockrelation_legal_info_end");
        spark.stop();*/
    }
}
